#include "execution/executors/topn_executor.h"

namespace bustub {

TopNExecutor::TopNExecutor(ExecutorContext *exec_ctx, const TopNPlanNode *plan,
                           std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      child_executor_(std::move(child_executor)),
      heap_(SortHelper(GetOutputSchema(), plan_->GetOrderBy())) {}

void TopNExecutor::Init() {
  // std::cout << plan_->ToString(true) << std::endl;
  child_executor_->Init();
  Tuple ch_tuple;
  RID ch_rid;
  heap_.clear();
  while (child_executor_->Next(&ch_tuple, &ch_rid)) {
    heap_.emplace(ch_tuple);
    if (heap_.size() > plan_->n_) {
      heap_.pop();
    }
  }
  top_n_.clear();
  top_n_.reserve(std::min(plan_->n_, heap_.size()));
  size_t i = 0;
  while (not heap_.empty()) {
    top_n_.push_back(heap_.top());
    heap_.pop();
    ++i;
  }
  std::reverse(top_n_.begin(), top_n_.end());
  cur_idx_ = 0;
}
auto TopNExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  while (cur_idx_ < top_n_.size()) {
    *tuple = top_n_[cur_idx_];
    *rid = tuple->GetRid();
    ++cur_idx_;
    return true;
  }
  return false;
}

auto TopNExecutor::GetNumInHeap() -> size_t { return heap_.size(); };

}  // namespace bustub
